//package com.dky.satell.kafka.acquisition;
//
//import com.dky.satell.entity.MessageConfig;
//import com.dky.satell.mapper.MessageConfigMapper;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.boot.ApplicationArguments;
//import org.springframework.boot.ApplicationRunner;
//import org.springframework.stereotype.Component;
//import org.springframework.util.CollectionUtils;
//import org.springframework.util.StringUtils;
//
//import javax.annotation.Resource;
//import java.util.HashMap;
//import java.util.List;
//import java.util.Map;
//
///**
// * kafka 采集消费者
// *
// * @author fagnyuan
// * @version 1.0
// * @date 2021/5/21 10:17
// */
//@Slf4j
//@Component
//public class KafkaConsumerInit implements ApplicationRunner {
//
//    @Resource
//    private MessageConfigMapper messageConfigMapper;
//
//    private Map<Integer, Consumer> consumerMap = new HashMap<>();
//
//    @Override
//    public void run(ApplicationArguments args) {
//        log.info("[start]初始化启动Kafka消费者！");
//        log.info("获取所有Kafka订阅信息！");
//        List<MessageConfig> messageConfigList = messageConfigMapper.getAllKafkaConfig();
//        if(CollectionUtils.isEmpty(messageConfigList)) {
//            return;
//        }
//        for(MessageConfig messageConfig : messageConfigList) {
//            try {
//                this.buildConsumer(messageConfig);
//            } catch (Exception e) {
//                log.error("消息队列[{}]订阅失败", messageConfig.getName(), e);
//            }
//        }
//        log.info("[end]初始化启动Kafka消费者！");
//    }
//
//    public void subscribe(MessageConfig messageConfig) {
//        try {
//            this.buildConsumer(messageConfig);
//        } catch (Exception e) {
//            log.error("消息队列[{}]订阅失败", messageConfig.getName(), e);
//        }
//    }
//
//    public void unsubscribe(MessageConfig messageConfig) {
//        Consumer consumer = consumerMap.get(messageConfig.getId());
//        if(consumer != null) {
////            consumer.unsubscribe();
//            consumer.interrupt();
//            consumerMap.remove(consumer);
//        }
//        log.info("取消订阅Kafka消费者[{}]！", messageConfig.getGroupId());
//    }
//
//    /**
//     * 初始化消费者
//     *
//     * @return
//     */
//    public Consumer getConsumer(Integer id) {
//        return consumerMap.get(id);
//    }
//
//    private boolean verifyConfig(MessageConfig messageConfig) {
//        if(!(StringUtils.hasLength(messageConfig.getNamesrvAddr()) && StringUtils.hasLength(messageConfig.getGroupId())
//                && StringUtils.hasLength(messageConfig.getTopic()))) {
//            log.error("消息队列[{}]Kafka配置信息有误", messageConfig.getName());
//            return false;
//        }
//        return true;
//    }
//
//    /**
//     * 构建单个消费者
//     */
//    private void buildConsumer(MessageConfig messageConfig) {
//        if(!this.verifyConfig(messageConfig)) {
//            return;
//        }
//        Consumer consumer = new Consumer(messageConfig);
//        consumer.start();
//        consumerMap.put(messageConfig.getId(), consumer);
//        log.info("初始化Kafka消费者[{}]！", messageConfig.getGroupId());
//    }
//}
